package com.wanglei.rdd.dependency

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Spark04_rddcheckpoint {

  def main(args: Array[String]): Unit = {
    // application
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("wc")
    val sc = new SparkContext(sparkConf)

    //
    val lines: RDD[String] = sc.textFile("datas/1.txt")
    val words: RDD[String] = lines.flatMap(_.split(" "))
    val wordToOne: RDD[(String, Int)] = words.map(word => (word, 1))

    // cache/persist 和 checkpoint的区别
    // cache/persist是临时存储，就是说当spark的application结束后，这部分数据会删除
    // checkpoint需要你指定路径，当app结束后，这个数据不会删除,注意cache和checkpoint一起使用
    // 因为checkpoint会新触发一个runjob，所以需要先cache，然后将内存的数据做checkpoint
    sc.setCheckpointDir("ck")
    wordToOne.cache()
    wordToOne.checkpoint()

    val res1: RDD[(String, Int)] = wordToOne.reduceByKey(_ + _)
    val res2: RDD[(String, Iterable[Int])] = wordToOne.groupByKey()


    // job1
    res1.collect()
    // job2
    res2.collect()
    //
    sc.stop()
  }

}
